package com.dongnao.rxjavademo;

import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.schedulers.Schedulers;

public class TestThreadActivity extends AppCompatActivity {

    private static final String TAG = TestThreadActivity.class.getName();

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_test_thread);
//        testThread();
//        testError();
//        testResumeError();

        testRetry();
    }


    public void testThread(){
        Observable.just(1,2).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                //子线程
                Log.e(TAG, "apply:----------->  "+ Thread.currentThread().getName());
                return integer +" ";
            }
        })./*subscribeOn(AndroidSchedulers.mainThread()).*/subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
               /* .observeOn(Schedulers.io())*/
                .subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                //主线程
                Log.e(TAG, "accept:----------->   "+ Thread.currentThread().getName());
            }
        });
    }


    //    适用于 可能会出现崩溃    但是结果不影响程序的运行。  比如说  访问数组时  数组越界 。
//        不适用 哪些程序本身就有错误的，而且这个错误会影响后面的，比如你要给用户展示一张图片，结果你从上一个页面传过来的url  就是空的。

    public void testError(){
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                for (int i = 0; i < 10; i++) {
                    if (i == 2){
                        e.onError(new Throwable("e can not be 2 !"));
                    }else {
                        e.onNext(i + "");
                    }
                }
            }
        }).onErrorReturn(new Function<Throwable, String>() {
            @Override
            public String apply(Throwable throwable) throws Exception {
                //当遇见异常，会调用该方法进行处理 ，替换掉 onError 方法 然后会调用 onNext方法
                Log.e(TAG, "apply: -----> " + throwable.getMessage() );
                return "参数错误";
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                Log.e(TAG, "onNext: -----> " + s );
            }

            @Override
            public void onError(Throwable e) {
                //此时 onError不会执行
                Log.e(TAG, "onError: -----> " + e.getMessage() );
            }

            @Override
            public void onComplete() {

            }
        });
    }

    public void testResumeError(){
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                for (int i = 0; i < 10; i++) {
                    if (i == 2){
                        e.onError(new Throwable("e can not be 2 !"));
                    }else {
                        e.onNext(i + "");
                    }
                }
            }
        }).onErrorResumeNext(new ObservableSource() {
            @Override
            public void subscribe(Observer observer) {
                //遇见异常，会创建一个新的事件即 Observable ，
                Log.e(TAG, "onErrorResumeNext: subscribe -----> " + observer );
                observer.onNext("错误修改");
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                Log.e(TAG, "onNext: -----> " + s );
            }

            @Override
            public void onError(Throwable e) {
                //此时 onError不会执行
                Log.e(TAG, "onError: -----> " + e.getMessage() );
            }

            @Override
            public void onComplete() {

            }
        });
    }


    public void testRetry(){
        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                for(int i = 0; i<= 9 ;i++){
                    if(i == 2){
//                        null
                        e.onError(new Exception("出现错误了"));
                    }else{
                        e.onNext(i+"");
                    }
                    try{
                        Thread.sleep(1000);
                    }catch (Exception ex){
                        ex.printStackTrace();
                    }
                }

                e.onComplete();
            }
        }).subscribeOn(Schedulers.newThread())
//        times  最多重复的次数
            .retry(3, new Predicate<Throwable>() {
                    @Override
                    public boolean test(Throwable throwable) throws Exception {
                        //返回false就是不让重新发射数据了，调用观察者的onError就终止了;返回true就是让被观察者重新发射请求
                        return true;
                    }
                })
            .subscribeOn(Schedulers.newThread()).subscribe(new Consumer<String>() {
                 @Override
                 public void accept(String s) throws Exception {

                 }
            },new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        Log.e(TAG, "结果错误: " + throwable.toString());
                    }
                });

    }



}
